package interceptor;

import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MySink extends AbstractSink implements Configurable {

    //创建logger对象
    private Logger logger = LoggerFactory.getLogger(MySink.class);

    //前后缀
    private String prefix;
    private String subfix;

    @Override
    public void configure(Context context) {
        prefix = context.getString("prefix");
        subfix = context.getString("subfix", "atguigu");
    }

    @Override
    public Status process() throws EventDeliveryException {
        Status status = null;
        //获取channel
        //从channel获取事务以及事件
        //发送数据
        Channel channel = getChannel();
        Transaction transaction = channel.getTransaction();
        transaction.begin();
        try {
            Event event = channel.take();
            if (event != null) {
                //处理获取到的事件（业务逻辑）
                String body = new String(event.getBody());
                logger.info(prefix + body + subfix);
            }
            //提交事务
            transaction.commit();
            status = Status.READY;
        } catch (ChannelException e) {
            e.printStackTrace();
            transaction.rollback();
            status = Status.BACKOFF;
        } finally {
            transaction.close();
        }
        return status;
    }


}
